package com.sjk.flink.functions;

import java.math.BigDecimal;
import java.util.Map;

import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.sjk.flink.analysis.IAnalysisHander;
import com.sjk.flink.util.JsonSchemaUtil;
import com.sjk.flink.util.ObjectMapperUtil;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

public class DorisProcessFunction
        extends ProcessFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>> {

    private static final Logger log = LoggerFactory.getLogger(DorisProcessFunction.class);

    private final OutputTag<String> valueTag;

    private final JsonNode fields;

    private final Map<String, IAnalysisHander> analysisMap;

    private final String type;

    public DorisProcessFunction(String type, JsonNode fields, Map<String, IAnalysisHander> analysisMap,
            OutputTag<String> valueTag) {
        this.valueTag = valueTag;
        this.fields = fields;
        this.analysisMap = analysisMap;
        this.type = type;
    }

    @Override
    public void processElement(ConsumerRecord<String, String> stringStringConsumerRecord,
            ProcessFunction<ConsumerRecord<String, String>, ConsumerRecord<String, String>>.Context ctx,
            Collector<ConsumerRecord<String, String>> collector) throws Exception {

        String value = stringStringConsumerRecord.value();
        log.info("processElement>>>         " + value);
        System.out.println("processElement>>>         " + value);

        try {
            ObjectMapper objectMapper = ObjectMapperUtil.getObjectMapper();
            // 城市分区
            int city_partition_id = 0;
            // 城市偏移量
            long city_offset_id = 0;
            // 时间戳
            String city_ts = "";
            // 华为分区
            int hw_partition_id = 0;
            // 华为偏移量
            long hw_offset_id = 0;
            String hw_ts = "";
            JsonNode readTree = objectMapper.readTree(value);
            JsonNode sscsdmJsonNode = readTree.get("sscsdm");
            // 城市代码
            String topicsscsdm = sscsdmJsonNode.asText();
            // 抽取时间
            // JsonNode extract_timeJsonNode = readTree.get("extract_time");
            // String extract_time = extract_timeJsonNode.asText();
            // 城市分区
            JsonNode city_partition_idJsonNode = readTree.get("city_partition_id");
            if (city_partition_idJsonNode != null) {
                city_partition_id = new Integer(city_partition_idJsonNode.asText());
            }
            // 城市偏移量
            JsonNode city_offset_idJsonNode = readTree.get("city_offset_id");
            if (city_offset_idJsonNode != null) {
                city_offset_id = new Long(city_offset_idJsonNode.asText());
            }
            // 时间戳
            JsonNode city_tsJsonNode = readTree.get("city_ts");
            if (city_tsJsonNode != null) {
                city_ts = city_tsJsonNode.asText();
            }

            // 华为分区
            JsonNode hw_partition_idJsonNode = readTree.get("hw_partition_id");
            if (hw_partition_idJsonNode != null) {
                hw_partition_id = new Integer(hw_partition_idJsonNode.asText());
            }
            // 华为偏移量
            JsonNode hw_offset_idJsonNode = readTree.get("hw_offset_id");
            if (hw_offset_idJsonNode != null) {
                hw_offset_id = new Long(hw_offset_idJsonNode.asText());
            }
            // 华为戳
            JsonNode hw_sJsonNode = readTree.get("hw_ts");
            if (hw_sJsonNode != null) {
                hw_ts = hw_sJsonNode.asText();
            }

            ObjectNode jsonObject = objectMapper.createObjectNode();
            jsonObject.put("sscsdm", topicsscsdm);
            if ("ODS".equals(type)) {
                jsonObject.put("record", value);
            }
            jsonObject.put("extract_time", JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));
            jsonObject.put("city_partiiton_id", city_partition_id);
            jsonObject.put("city_offset_id", city_offset_id);
            jsonObject.put("city_ts", city_ts);
            jsonObject.put("hw_partition_id", hw_partition_id);
            jsonObject.put("hw_offset_id", hw_offset_id);
            jsonObject.put("hw_ts", hw_ts);
            if (fields != null && fields.isArray()) {
                for (JsonNode node1 : fields) {
                    JsonNode originNode = node1.get("origin");
                    JsonNode toNode = node1.get("to");
                    JsonNode analysisHanderNode = node1.get("analysisHander");
                    if (analysisHanderNode != null) {
                        IAnalysisHander hander = analysisMap.get(analysisHanderNode.asText());
                        if (hander != null && toNode != null) {
                            String to = toNode.asText();
                            Object val = hander.doAnalysis(readTree);
                            if (val instanceof Integer) {
                                jsonObject.put(to, (Integer) val);
                            } else if (val instanceof String) {
                                jsonObject.put(to, (String) val);
                            } else if (val instanceof Double) {
                                jsonObject.put(to, (Double) val);
                            } else if (val instanceof Float) {
                                jsonObject.put(to, (Float) val);
                            } else if (val instanceof Long) {
                                jsonObject.put(to, (Long) val);
                            } else if (val instanceof Boolean) {
                                jsonObject.put(to, (Boolean) val);
                            } else if (val instanceof BigDecimal) {
                                jsonObject.put(to, (BigDecimal) val);
                            } else {
                                // 默认设置为NULL
                                jsonObject.putNull(to);
                            }
                        }
                    } else {
                        // 没有解析
                        if (originNode != null && toNode != null) {
                            String origin = originNode.asText();
                            String to = toNode.asText();
                            JsonNode jsonNodeVal = readTree.get(origin);
                            if (jsonNodeVal == null) {
                                jsonObject.put(to, "");
                            } else {
                                jsonObject.put(to, jsonNodeVal.asText());
                            }
                        }
                    }

                }
            }
            String outJosn = objectMapper.writeValueAsString(jsonObject);
            log.info("输出到Doris数据>>>         " + outJosn);
            System.out.println("输出到Doris数据>>>         " + outJosn);
            // 添加
            ctx.output(valueTag, outJosn);
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("解析出错:   "+ ex.getMessage());
            log.error(value, ex);
        }
    }

}
